package com.example.webbackend.service.impl;

import com.example.webbackend.config.RabbitMQConfig;
import com.example.webbackend.listener.MessageConsumer;
import com.example.webbackend.model.Message;
import com.example.webbackend.model.OfflineMessage;
import com.example.webbackend.repository.MessageRepository;
import com.example.webbackend.repository.OfflineMessageRepository;
import com.example.webbackend.service.MessageService;
import org.redisson.api.RedissonClient;
import org.redisson.api.RList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;


@Service
public class MessageServiceImpl implements MessageService {

    private static final Logger logger = LoggerFactory.getLogger(MessageConsumer.class);
    private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>();
    private final MessageRepository messageRepository;
    private final OfflineMessageRepository offlineMessageRepository;
    private final RedissonClient redissonClient;
    private final RabbitTemplate rabbitTemplate;
    private static final String MESSAGE_CACHE_PREFIX = "msg:";

    public MessageServiceImpl(MessageRepository messageRepository,
                              OfflineMessageRepository offlineMessageRepository,
                              RedissonClient redissonClient,
                              RabbitTemplate rabbitTemplate) {
        this.messageRepository = messageRepository;
        this.offlineMessageRepository = offlineMessageRepository;
        this.redissonClient = redissonClient;
        this.rabbitTemplate = rabbitTemplate;
        logger.info("RabbitMQ sender enabled: MessageServiceImpl is ready to send messages.");
    }

    @Override
    public void cacheMessage(Message message) {
        String key = MESSAGE_CACHE_PREFIX + message.getReceiver();
        RList<Message> msgList = redissonClient.getList(key);
        msgList.add(message); // 分布式落盘，自动分片
        // 可以设置过期策略：例如只保留最新的1000条
        if (msgList.size() > 1000) {
            msgList.remove(0); // 保持容量
        }
    }

    @Override
    public List<Message> getCachedMessages(String username) {
        String key = MESSAGE_CACHE_PREFIX + username;
        RList<Message> msgList = redissonClient.getList(key);
        // 这里直接返回全部消息，也可只返回最近N条
        return msgList.readAll();
    }

    @Override
    public void clearMessageCache(String username) {
        String key = MESSAGE_CACHE_PREFIX + username;
        redissonClient.getList(key).delete();
    }

    @Override
    public SseEmitter subscribe(String username) {
        SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30分钟超时
        emitters.put(username, emitter);

        emitter.onCompletion(() -> emitters.remove(username));
        emitter.onTimeout(() -> emitters.remove(username));

        // 修改为从数据库获取离线消息
        List<Message> messages = getOfflineMessages(username);
        if (!messages.isEmpty()) {
            try {
                for (Message message : messages) {
                    emitter.send(message);
                }
            } catch (Exception e) {
                emitter.completeWithError(e);
            }
        }
        return emitter;
    }
    @Override
    public void processAndPushMessage(Message message) {
        cacheMessage(message);
        if (emitters.containsKey(message.getReceiver())) {
            try {
                message.setStatus("DELIVERED");
                emitters.get(message.getReceiver()).send(message);
                messageRepository.save(message);
            } catch (Exception e) {
                emitters.remove(message.getReceiver());
            }
        } else {
            OfflineMessage offlineMsg = new OfflineMessage();
            offlineMsg.setMessageId(message.getId());
            offlineMsg.setReceiver(message.getReceiver());
            offlineMessageRepository.save(offlineMsg);
        }
    }

    @Override
    public Message sendMessage(Message message) {
        message.setStatus("SENT");
        message.setCreateTime(LocalDateTime.now());
        messageRepository.save(message);
        logger.debug("[RabbitMQ] Sending message to queue: {}", message);
        rabbitTemplate.convertAndSend(RabbitMQConfig.MESSAGE_QUEUE, message);
        return message;
    }


    @Override
    public List<Message> getMessageHistory(Long groupId) {
        return messageRepository.findByGroupIdOrderByCreateTimeAsc(groupId);
    }

    @Override
    public List<Message> getOfflineMessages(String username) {
        // 先检查分布式缓存
        List<Message> cached = getCachedMessages(username);
        if (!cached.isEmpty()) {
            clearMessageCache(username); // 离线消息发出后及时清理
            return cached;
        }

        // 先获取7天内未过期的离线消息
        List<Long> messageIds = offlineMessageRepository.findValidMessagesByReceiver(username,
                        LocalDateTime.now().minusDays(7))
                .stream()
                .map(OfflineMessage::getMessageId)
                .collect(Collectors.toList());

        if (messageIds.isEmpty()) {
            return List.of();
        }

        List<Message> messages = messageRepository.findAllById(messageIds);
        offlineMessageRepository.deleteByReceiver(username);

        // 批量更新消息状态为已读
        messages.forEach(msg -> msg.setStatus("READ"));
        messageRepository.saveAll(messages);

        return messages;
    }
}
